package com.shujia.flink

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08


object Demo9ReadKafka {
  def main(args: Array[String]): Unit = {

    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    //设置时间类型为事件事件
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val tableEnv = StreamTableEnvironment.create(streamEnv)


    // configure Kafka consumer
    val kafkaProps = new Properties()
    kafkaProps.setProperty("zookeeper.connect", "node1:2181,node2:2181,node3:2181")
    kafkaProps.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092")
    kafkaProps.setProperty("group.id", "asdasd")
    //    kafkaProps.setProperty("auto.offset.reset", "earliest")
    //    kafkaProps.setProperty("key.deserializer", "kafka.serializer.StringDecoder")


    //构建kafka消费者
    val consumer = new FlinkKafkaConsumer08[String](
      "topic",
      new SimpleStringSchema(),
      kafkaProps)

    //设置从哪里消费数据
    consumer.setStartFromEarliest() //从最早消费数据


    //构建kafkaDS
    val kafkaSource = streamEnv.addSource(consumer)

    kafkaSource.print()


    streamEnv.execute()

  }
}
